package org.aurora.remote.netty;

import io.netty.channel.*;
import org.aurora.common.delay.queue.DelayExecuteTaskQueue;
import org.aurora.common.delay.queue.DelayTask;
import org.aurora.common.serialization.Byte2ObjectSerializationFactory;
import org.aurora.remote.*;
import org.aurora.remote.exception.RequestTimeoutException;
import org.aurora.remote.exception.SendRequestFailException;
import org.aurora.remote.exception.SendRequestInterruptedException;
import org.aurora.remote.protocol.RemoteCommand;
import org.aurora.remote.protocol.RemoteRequest;
import org.aurora.remote.protocol.RemoteResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@ChannelHandler.Sharable
public class NettyRemoteCommandHandler extends SimpleChannelInboundHandler<RemoteCommand> implements RemoteCommandHandler<NettyRemoteCommandContext> {
    private static final Logger log = LoggerFactory.getLogger(NettyRemoteCommandHandler.class);

    private final Map<Integer, RemoteProcessor<RemoteRequest, RemoteResponse>> requestProcessors = new ConcurrentHashMap<>();
    private final Map<Integer, ResponseFuture> responseTable = new ConcurrentHashMap<>();
    private DelayExecuteTaskQueue responseDelayQueue;
    private NettyEventPublisher nettyEventPublisher;
    private NettyConnectChangeHandler nettyConnectChangeHandler;

    public NettyRemoteCommandHandler() {
        this.nettyEventPublisher = new NettyEventPublisher();
        this.nettyConnectChangeHandler = new NettyConnectChangeHandler(this.nettyEventPublisher);
    }

    public NettyConnectChangeHandler getNettyConnectChangeHandler() {
        return nettyConnectChangeHandler;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RemoteCommand remoteCommand) throws Exception {
        Channel channel = ctx.channel();
        fillNetworkParam(channel, remoteCommand);

        processor(new NettyRemoteCommandContext(channel, remoteCommand));
    }

    private void fillNetworkParam(Channel channel, RemoteCommand remoteCommand) {
        InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
        String ip = inetSocketAddress.getAddress().getHostAddress();
        int port = inetSocketAddress.getPort();
        remoteCommand.setIp(ip);
        remoteCommand.setPort(port);
    }

    @Override
    public void processorRequest(NettyRemoteCommandContext context) {
        RemoteRequest request = context.getRemoteRequest();
        Channel channel = context.getChannel();

        int commandCode = request.getCommandCode();

        RemoteProcessor<RemoteRequest, RemoteResponse> remoteProcessor = requestProcessors.get(commandCode);

        if (remoteProcessor == null) {
            throw new RuntimeException("不能处理的请求");
        }

        // todo 考虑是否使用多线程
        // todo 考虑是否处理无响应请求
        RemoteResponse remoteResponse = remoteProcessor.processor(request);
        remoteResponse.setRequestOrdinal(request.getRequestOrdinal());

        if (remoteResponse != null) {
            channel.writeAndFlush(remoteResponse);
        }
    }

    @Override
    public void processorResponse(NettyRemoteCommandContext context) {
        RemoteResponse response = context.getRemoteResponse();
        ResponseFuture responseFuture = responseTable.remove(response.getRequestOrdinal());

        if (responseFuture == null) {
            return;
        }
        // 同步 发送完以后等待，从 responseTable 中拿数据

        WaitResponseType waitResponseType = responseFuture.getWaitResponseType();

        switch (waitResponseType) {
            case SYNC: {
                responseFuture.putResponse(response);
                break;
            }
            case ASYNC: {
                responseFuture.putResponse(response);
                SendCallback sendCallback = responseFuture.getSendCallback();

                sendCallback.onSuccess((RemoteResponse) Byte2ObjectSerializationFactory.deserialization(response.getData(), sendCallback.getResponseClass(), response.getSerializationType()));
                break;
            }
        }
    }

    @Override
    public void start() {
        this.responseDelayQueue = new DelayExecuteTaskQueue("nettyRemoteCommandHandlerResponseDelayQueue", new DelayExecuteTaskQueue.DelayExecuteTaskActuator<AsyncSendTimeoutCleanTask>() {
            @Override
            public void execute(AsyncSendTimeoutCleanTask delayTask) {
                ResponseFuture responseFuture = delayTask.getResponseFuture();

                if (responseFuture == null) {
                    return;
                }

                RemoteRequest request = responseFuture.getRequest();

                if (request == null) {
                    return;
                }

                responseFuture = responseTable.remove(request.getRequestOrdinal());

                if (responseFuture != null) {
                    SendCallback sendCallback = responseFuture.getSendCallback();

                    if (sendCallback != null) {
                        sendCallback.onFail(new RuntimeException("timeout"));
                    }
                }
            }
        });

        this.responseDelayQueue.start();
        this.nettyEventPublisher.start();
    }

    @Override
    public void stop() {
        this.responseDelayQueue.stop();
    }

    /**
     *
     * @param commandCode
     * @param processor
     */
    public void registerProcessor(int commandCode, RemoteProcessor<RemoteRequest, RemoteResponse> processor) {
        requestProcessors.put(commandCode, processor);
    }

    public void registerEventSubscriber(NettyEventSubscriber nettyEventSubscriber) {
        this.nettyEventPublisher.addSubscriber(nettyEventSubscriber);
    }

    public RemoteResponse send(Channel channel, RemoteRequest request, long timeoutMillis) {
        ChannelFuture channelFuture = null;

        request.setRequestOrdinal(RemoteRequest.requestOrdinalGeneral());
        ResponseFuture responseFuture = new ResponseFuture(WaitResponseType.SYNC);
        responseTable.put(request.getRequestOrdinal(), responseFuture);
        try {
            channelFuture = channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        responseFuture.requestSuccess();
                        return;
                    }

                    responseFuture.requestFail();
                }
            });
        } catch (Exception e) {
            log.error("send message error", e);
            throw new SendRequestFailException(e);
        }

        RemoteResponse response = null;
        try {
            response = responseFuture.waitResponse(timeoutMillis);

            if (response == null) {
                responseTable.remove(request.getRequestOrdinal());

                if (responseFuture.isRequestSuccess()) {
                    log.error("send message timeout");
                    throw new RequestTimeoutException();
                } else {
                    log.error("send message error");
                    throw new SendRequestFailException();
                }
            }
        } catch (InterruptedException e) {
            log.error("send message interrupted error", e);
            throw new SendRequestInterruptedException("send message interrupted", e);
        }

        // 不为空说明已经获取到正常的响应不需要从 responseTable 中移除

        return response;
    }

    public void sendAsync(Channel channel, RemoteRequest request, long timeoutMillis, SendCallback sendCallback) {
        request.setRequestOrdinal(RemoteRequest.requestOrdinalGeneral());
        ResponseFuture responseFuture = new ResponseFuture(WaitResponseType.ASYNC, request, sendCallback);

        responseTable.put(request.getRequestOrdinal(), responseFuture);
        responseDelayQueue.put(new AsyncSendTimeoutCleanTask(timeoutMillis, responseFuture));
        try {
            channel.writeAndFlush(request);
        } catch (Exception e) {
            log.error("send message error", e);
            throw new SendRequestFailException(e);
        }
    }

    /**
     * 异步请求的清理任务
     */
    private class AsyncSendTimeoutCleanTask extends DelayTask {

        private ResponseFuture responseFuture;

        public AsyncSendTimeoutCleanTask(long delayTimeMillis, ResponseFuture responseFuture) {
            super(delayTimeMillis);
            this.responseFuture = responseFuture;
        }

        public ResponseFuture getResponseFuture() {
            return responseFuture;
        }

        public void setResponseFuture(ResponseFuture responseFuture) {
            this.responseFuture = responseFuture;
        }
    }

}
